//! 基于消息总线的配置广播抽象实现
//! 
//! 通过定义抽象接口避免与 rustcloud-bus 的循环依赖
//! 具体的消息总线实现由外部注入

use crate::broadcast::*;
use crate::error::{ConfigResult, ConfigError};
use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{RwLock, Mutex};
use tracing::{info, error, debug, warn};
use serde_json::Value;

/// 抽象消息接口 - 避免直接依赖 rustcloud-bus
#[derive(Debug, Clone)]
pub struct ConfigMessage {
    /// 消息ID
    pub id: String,
    /// 主题
    pub topic: String,
    /// 消息内容
    pub payload: Value,
    /// 时间戳
    pub timestamp: std::time::SystemTime,
    /// 消息头
    pub headers: HashMap<String, String>,
    /// 关联ID
    pub correlation_id: Option<String>,
}

impl ConfigMessage {
    /// 创建新消息
    pub fn new(topic: String, payload: Value) -> Self {
        Self {
            id: format!("msg_{}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos()),
            topic,
            payload,
            timestamp: std::time::SystemTime::now(),
            headers: HashMap::new(),
            correlation_id: None,
        }
    }
    
    /// 添加消息头
    pub fn with_header(mut self, key: String, value: String) -> Self {
        self.headers.insert(key, value);
        self
    }
    
    /// 设置关联ID
    pub fn with_correlation_id(mut self, correlation_id: String) -> Self {
        self.correlation_id = Some(correlation_id);
        self
    }
}

/// 抽象消息发布者接口
#[async_trait]
pub trait ConfigMessagePublisher: Send + Sync {
    /// 发布消息
    async fn publish(&self, message: ConfigMessage) -> Result<(), String>;
}

/// 抽象消息订阅者接口
#[async_trait]
pub trait ConfigMessageSubscriber: Send + Sync {
    /// 订阅主题
    async fn subscribe(&self, topic: &str) -> Result<(), String>;
    
    /// 接收消息
    async fn receive(&self) -> Result<Option<ConfigMessage>, String>;
    
    /// 取消订阅
    async fn unsubscribe(&self, topic: &str) -> Result<(), String>;
}

/// 抽象消息总线接口
#[async_trait]
pub trait ConfigMessageBus: Send + Sync {
    /// 创建发布者
    async fn create_publisher(&self) -> Result<Box<dyn ConfigMessagePublisher>, String>;
    
    /// 创建订阅者
    async fn create_subscriber(&self) -> Result<Box<dyn ConfigMessageSubscriber>, String>;
}

/// 基于抽象消息总线的配置广播管理器
pub struct AbstractConfigBroadcastManager {
    /// 抽象消息总线实例
    message_bus: Arc<dyn ConfigMessageBus>,
    /// 服务名称
    service_name: String,
    /// 广播主题前缀
    topic_prefix: String,
    /// 配置变更监听器
    listeners: Arc<RwLock<HashMap<String, Box<dyn ConfigBroadcastListener>>>>,
    /// 集群状态缓存
    cluster_state: Arc<RwLock<HashMap<String, ServiceState>>>,
    /// 广播统计信息
    stats: Arc<Mutex<BroadcastStats>>,
    /// 是否正在运行
    running: Arc<RwLock<bool>>,
    /// 消息订阅者
    subscriber: Arc<Mutex<Option<Box<dyn ConfigMessageSubscriber>>>>,
}

impl AbstractConfigBroadcastManager {
    /// 创建新的配置广播管理器
    pub fn new(
        message_bus: Arc<dyn ConfigMessageBus>,
        service_name: String,
        topic_prefix: Option<String>,
    ) -> Self {
        Self {
            message_bus,
            service_name,
            topic_prefix: topic_prefix.unwrap_or_else(|| "config.broadcast".to_string()),
            listeners: Arc::new(RwLock::new(HashMap::new())),
            cluster_state: Arc::new(RwLock::new(HashMap::new())),
            stats: Arc::new(Mutex::new(BroadcastStats::default())),
            running: Arc::new(RwLock::new(false)),
            subscriber: Arc::new(Mutex::new(None)),
        }
    }
    
    /// 静态方法处理消息（解决生命周期问题）
    async fn handle_message_static(
        message: &ConfigMessage,
        listeners: &Arc<RwLock<HashMap<String, Box<dyn ConfigBroadcastListener>>>>,
        cluster_state: &Arc<RwLock<HashMap<String, ServiceState>>>,
        stats: &Arc<Mutex<BroadcastStats>>,
        service_name: &str,
    ) -> ConfigResult<()> {
        // 解析配置广播事件
        let event: ConfigBroadcastEvent = serde_json::from_value(message.payload.clone())
            .map_err(|e| ConfigError::ParseError(format!("解析广播事件失败: {}", e)))?;
        
        // 跳过自己发送的事件
        if event.source_service == service_name {
            debug!("跳过自己发送的事件: {}", event.id);
            return Ok(());
        }
        
        // 更新统计信息
        {
            let mut stats_guard = stats.lock().await;
            stats_guard.events_received += 1;
        }
        
        // 检查是否是目标服务
        if let Some(targets) = &event.target_services {
            if !targets.contains(&service_name.to_string()) {
                debug!("事件不针对当前服务: {} -> {:?}", service_name, targets);
                return Ok(());
            }
        }
        
        info!(
            "收到配置广播事件: {} -> {} ({})",
            event.event_type.as_str(),
            event.application,
            event.id
        );
        
        // 更新集群状态
        Self::update_cluster_state_static(&event, cluster_state).await;
        
        // 分发给所有关注的监听器
        let listeners_guard = listeners.read().await;
        let mut processed_count = 0;
        let mut failed_count = 0;
        
        for (name, listener) in listeners_guard.iter() {
            if Self::should_listener_handle_static(listener.as_ref(), &event) {
                match listener.on_config_change(&event).await {
                    Ok(()) => {
                        processed_count += 1;
                        debug!("监听器 {} 处理事件成功", name);
                    }
                    Err(e) => {
                        failed_count += 1;
                        error!("监听器 {} 处理事件失败: {}", name, e);
                    }
                }
            }
        }
        
        // 更新统计信息
        {
            let mut stats_guard = stats.lock().await;
            stats_guard.events_processed += processed_count;
            stats_guard.events_failed += failed_count;
            stats_guard.last_processed = Some(std::time::SystemTime::now());
        }
        
        info!(
            "配置广播事件处理完成: 成功={}, 失败={}",
            processed_count, failed_count
        );
        
        Ok(())
    }
    
    /// 静态方法判断监听器是否应该处理此事件
    fn should_listener_handle_static(
        listener: &dyn ConfigBroadcastListener,
        event: &ConfigBroadcastEvent,
    ) -> bool {
        // 基本过滤
        if !listener.should_handle(event) {
            return false;
        }
        
        // 应用过滤
        let interested_apps = listener.interested_applications();
        if !interested_apps.is_empty() && !interested_apps.contains(&event.application) {
            return false;
        }
        
        true
    }
    
    /// 静态方法更新集群状态
    async fn update_cluster_state_static(
        event: &ConfigBroadcastEvent,
        cluster_state: &Arc<RwLock<HashMap<String, ServiceState>>>,
    ) {
        let mut cluster_state_guard = cluster_state.write().await;
        
        let state = cluster_state_guard
            .entry(event.source_service.clone())
            .or_insert_with(|| ServiceState {
                service_name: event.source_service.clone(),
                last_heartbeat: std::time::SystemTime::now(),
                config_version: "unknown".to_string(),
                status: ServiceStatus::Running,
                metadata: HashMap::new(),
            });
        
        state.last_heartbeat = std::time::SystemTime::now();
        
        // 根据事件类型更新状态
        match event.event_type {
            ConfigBroadcastEventType::ServiceRestart => {
                state.status = ServiceStatus::Restarting;
            }
            ConfigBroadcastEventType::ConfigUpdate | ConfigBroadcastEventType::ConfigRefresh => {
                state.status = ServiceStatus::Syncing;
            }
            _ => {
                state.status = ServiceStatus::Running;
            }
        }
    }

    /// 处理收到的配置广播消息
    async fn handle_broadcast_message(&self, message: &ConfigMessage) -> ConfigResult<()> {
        // 解析配置广播事件
        let event: ConfigBroadcastEvent = serde_json::from_value(message.payload.clone())
            .map_err(|e| ConfigError::ParseError(format!("解析广播事件失败: {}", e)))?;
        
        // 跳过自己发送的事件
        if event.source_service == self.service_name {
            debug!("跳过自己发送的事件: {}", event.id);
            return Ok(());
        }
        
        // 更新统计信息
        {
            let mut stats = self.stats.lock().await;
            stats.events_received += 1;
        }
        
        // 检查是否是目标服务
        if let Some(targets) = &event.target_services {
            if !targets.contains(&self.service_name) {
                debug!("事件不针对当前服务: {} -> {:?}", self.service_name, targets);
                return Ok(());
            }
        }
        
        info!(
            "收到配置广播事件: {} -> {} ({})",
            event.event_type.as_str(),
            event.application,
            event.id
        );
        
        // 更新集群状态
        self.update_cluster_state(&event).await;
        
        // 分发给所有关注的监听器
        let listeners = self.listeners.read().await;
        let mut processed_count = 0;
        let mut failed_count = 0;
        
        for (name, listener) in listeners.iter() {
            if self.should_listener_handle(listener.as_ref(), &event) {
                match listener.on_config_change(&event).await {
                    Ok(()) => {
                        processed_count += 1;
                        debug!("监听器 {} 处理事件成功", name);
                    }
                    Err(e) => {
                        failed_count += 1;
                        error!("监听器 {} 处理事件失败: {}", name, e);
                    }
                }
            }
        }
        
        // 更新统计信息
        {
            let mut stats = self.stats.lock().await;
            stats.events_processed += processed_count;
            stats.events_failed += failed_count;
            stats.last_processed = Some(std::time::SystemTime::now());
        }
        
        info!(
            "配置广播事件处理完成: 成功={}, 失败={}",
            processed_count, failed_count
        );
        
        Ok(())
    }
    
    /// 判断监听器是否应该处理此事件
    fn should_listener_handle(
        &self,
        listener: &dyn ConfigBroadcastListener,
        event: &ConfigBroadcastEvent,
    ) -> bool {
        // 基本过滤
        if !listener.should_handle(event) {
            return false;
        }
        
        // 应用过滤
        let interested_apps = listener.interested_applications();
        if !interested_apps.is_empty() && !interested_apps.contains(&event.application) {
            return false;
        }
        
        true
    }
    
    /// 更新集群状态
    async fn update_cluster_state(&self, event: &ConfigBroadcastEvent) {
        let mut cluster_state = self.cluster_state.write().await;
        
        let state = cluster_state
            .entry(event.source_service.clone())
            .or_insert_with(|| ServiceState {
                service_name: event.source_service.clone(),
                last_heartbeat: std::time::SystemTime::now(),
                config_version: "unknown".to_string(),
                status: ServiceStatus::Running,
                metadata: HashMap::new(),
            });
        
        state.last_heartbeat = std::time::SystemTime::now();
        
        // 根据事件类型更新状态
        match event.event_type {
            ConfigBroadcastEventType::ServiceRestart => {
                state.status = ServiceStatus::Restarting;
            }
            ConfigBroadcastEventType::ConfigUpdate | ConfigBroadcastEventType::ConfigRefresh => {
                state.status = ServiceStatus::Syncing;
            }
            _ => {
                state.status = ServiceStatus::Running;
            }
        }
    }
    
    /// 广播事件通用方法
    async fn broadcast_event(&self, event: ConfigBroadcastEvent) -> ConfigResult<()> {
        let topic = format!("{}.{}", self.topic_prefix, event.event_type.as_str());
        
        let message = ConfigMessage::new(topic, serde_json::to_value(&event)
            .map_err(|e| ConfigError::ParseError(format!("广播事件序列化失败: {}", e)))?)
            .with_header("event-type".to_string(), event.event_type.as_str().to_string())
            .with_header("source-service".to_string(), event.source_service.clone())
            .with_header("application".to_string(), event.application.clone())
            .with_correlation_id(event.id.clone());
        
        let publisher = self.message_bus.create_publisher().await
            .map_err(|e| ConfigError::NetworkError(format!("创建发布者失败: {}", e)))?;
        
        publisher.publish(message).await
            .map_err(|e| ConfigError::NetworkError(format!("发布消息失败: {}", e)))?;
        
        // 更新统计信息
        {
            let mut stats = self.stats.lock().await;
            stats.events_sent += 1;
        }
        
        info!(
            "配置广播事件已发送: {} -> {} ({})",
            event.event_type.as_str(),
            event.application,
            event.id
        );
        
        Ok(())
    }
    
    /// 启动消息接收循环
    async fn start_message_loop(&self) -> ConfigResult<()> {
        let subscriber = self.message_bus.create_subscriber().await
            .map_err(|e| ConfigError::NetworkError(format!("创建订阅者失败: {}", e)))?;
        
        // 订阅所有配置广播主题
        let topics = vec![
            format!("{}.config.update", self.topic_prefix),
            format!("{}.config.delete", self.topic_prefix),
            format!("{}.config.refresh", self.topic_prefix),
            format!("{}.service.restart", self.topic_prefix),
        ];
        
        for topic in &topics {
            subscriber.subscribe(topic).await
                .map_err(|e| ConfigError::NetworkError(format!("订阅主题失败 {}: {}", topic, e)))?;
        }
        
        // 保存订阅者引用
        {
            let mut sub_guard = self.subscriber.lock().await;
            *sub_guard = Some(subscriber);
        }
        
        // 启动消息处理循环
        let running = self.running.clone();
        let subscriber_ref = self.subscriber.clone();
        let listeners = self.listeners.clone();
        let cluster_state = self.cluster_state.clone();
        let stats = self.stats.clone();
        let service_name = self.service_name.clone();
        
        tokio::spawn(async move {
            loop {
                // 检查是否还在运行
                {
                    let is_running = *running.read().await;
                    if !is_running {
                        break;
                    }
                }
                
                // 接收消息
                let message = {
                    let subscriber_guard = subscriber_ref.lock().await;
                    if let Some(ref subscriber) = *subscriber_guard {
                        subscriber.receive().await
                    } else {
                        continue;
                    }
                };
                
                match message {
                    Ok(Some(msg)) => {
                        if let Err(e) = Self::handle_message_static(
                            &msg, &listeners, &cluster_state, &stats, &service_name
                        ).await {
                            error!("处理广播消息失败: {}", e);
                        }
                    }
                    Ok(None) => {
                        // 没有消息，稍等片刻
                        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                    }
                    Err(e) => {
                        error!("接收消息失败: {}", e);
                        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                    }
                }
            }
            
            info!("配置广播消息循环已停止");
        });
        
        Ok(())
    }
}

#[async_trait]
impl ConfigBroadcaster for AbstractConfigBroadcastManager {
    async fn broadcast_config_update(
        &self,
        application: &str,
        profile: &str,
        label: Option<&str>,
        config_path: &str,
        old_value: Option<serde_json::Value>,
        new_value: serde_json::Value,
        user: Option<&str>,
    ) -> ConfigResult<()> {
        let event = ConfigBroadcastEvent::config_update(
            application,
            profile,
            label,
            config_path,
            old_value,
            new_value,
            &self.service_name,
            user,
        );
        
        self.broadcast_event(event).await
    }
    
    async fn broadcast_config_delete(
        &self,
        application: &str,
        profile: &str,
        label: Option<&str>,
        config_path: &str,
        user: Option<&str>,
    ) -> ConfigResult<()> {
        let mut event = ConfigBroadcastEvent::config_update(
            application,
            profile,
            label,
            config_path,
            None,
            serde_json::Value::Null,
            &self.service_name,
            user,
        );
        event.event_type = ConfigBroadcastEventType::ConfigDelete;
        
        self.broadcast_event(event).await
    }
    
    async fn broadcast_config_refresh(
        &self,
        application: &str,
        profile: &str,
        target_services: Option<Vec<String>>,
    ) -> ConfigResult<()> {
        let event = ConfigBroadcastEvent::config_refresh(
            application,
            profile,
            &self.service_name,
            target_services,
        );
        
        self.broadcast_event(event).await
    }
    
    async fn broadcast_service_restart(
        &self,
        application: &str,
        target_services: Option<Vec<String>>,
        user: Option<&str>,
    ) -> ConfigResult<()> {
        let event = ConfigBroadcastEvent::service_restart(
            application,
            &self.service_name,
            target_services,
            user,
        );
        
        self.broadcast_event(event).await
    }
}

#[async_trait]
impl ConfigBroadcastManager for AbstractConfigBroadcastManager {
    async fn start(&self) -> ConfigResult<()> {
        {
            let mut running = self.running.write().await;
            if *running {
                return Err(ConfigError::ValidationError("配置广播管理器已在运行".to_string()));
            }
            *running = true;
        }
        
        info!("启动配置广播管理器: {}", self.service_name);
        
        // 启动消息接收循环
        self.start_message_loop().await?;
        
        info!("配置广播管理器启动成功");
        Ok(())
    }
    
    async fn stop(&self) -> ConfigResult<()> {
        {
            let mut running = self.running.write().await;
            if !*running {
                return Ok(());
            }
            *running = false;
        }
        
        info!("停止配置广播管理器: {}", self.service_name);
        
        // 取消订阅
        if let Some(subscriber) = self.subscriber.lock().await.as_ref() {
            let topics = vec![
                format!("{}.config.update", self.topic_prefix),
                format!("{}.config.delete", self.topic_prefix),
                format!("{}.config.refresh", self.topic_prefix),
                format!("{}.service.restart", self.topic_prefix),
            ];
            
            for topic in &topics {
                if let Err(e) = subscriber.unsubscribe(topic).await {
                    warn!("取消订阅主题失败 {}: {}", topic, e);
                }
            }
        }
        
        info!("配置广播管理器已停止");
        Ok(())
    }
    
    async fn register_listener(&self, listener: Box<dyn ConfigBroadcastListener>) {
        let name = listener.name().to_string();
        let mut listeners = self.listeners.write().await;
        listeners.insert(name.clone(), listener);
        
        // 更新统计信息
        {
            let mut stats = self.stats.lock().await;
            stats.active_listeners = listeners.len() as u32;
        }
        
        info!("注册配置广播监听器: {}", name);
    }
    
    async fn remove_listener(&self, name: &str) -> bool {
        let mut listeners = self.listeners.write().await;
        let removed = listeners.remove(name).is_some();
        
        if removed {
            // 更新统计信息
            {
                let mut stats = self.stats.lock().await;
                stats.active_listeners = listeners.len() as u32;
            }
            
            info!("移除配置广播监听器: {}", name);
        }
        
        removed
    }
    
    async fn get_cluster_state(&self) -> HashMap<String, ServiceState> {
        let cluster_state = self.cluster_state.read().await;
        cluster_state.clone()
    }
    
    async fn get_broadcast_stats(&self) -> BroadcastStats {
        let stats = self.stats.lock().await;
        stats.clone()
    }
}

/// 内存实现的消息总线（用于测试和演示）
pub struct InMemoryConfigMessageBus {
    message_queue: Arc<RwLock<Vec<ConfigMessage>>>,
}

impl InMemoryConfigMessageBus {
    pub fn new() -> Self {
        Self {
            message_queue: Arc::new(RwLock::new(Vec::new())),
        }
    }
}

#[async_trait]
impl ConfigMessageBus for InMemoryConfigMessageBus {
    async fn create_publisher(&self) -> Result<Box<dyn ConfigMessagePublisher>, String> {
        let publisher = InMemoryPublisher {
            message_queue: self.message_queue.clone(),
        };
        
        Ok(Box::new(publisher))
    }
    
    async fn create_subscriber(&self) -> Result<Box<dyn ConfigMessageSubscriber>, String> {
        let subscriber = InMemorySubscriber {
            message_queue: self.message_queue.clone(),
            subscribed_topics: Arc::new(RwLock::new(Vec::new())),
            last_read_index: Arc::new(Mutex::new(0)),
        };
        
        Ok(Box::new(subscriber))
    }
}

struct InMemoryPublisher {
    message_queue: Arc<RwLock<Vec<ConfigMessage>>>,
}

#[async_trait]
impl ConfigMessagePublisher for InMemoryPublisher {
    async fn publish(&self, message: ConfigMessage) -> Result<(), String> {
        let mut queue = self.message_queue.write().await;
        queue.push(message);
        Ok(())
    }
}

struct InMemorySubscriber {
    message_queue: Arc<RwLock<Vec<ConfigMessage>>>,
    subscribed_topics: Arc<RwLock<Vec<String>>>,
    last_read_index: Arc<Mutex<usize>>,
}

#[async_trait]
impl ConfigMessageSubscriber for InMemorySubscriber {
    async fn subscribe(&self, topic: &str) -> Result<(), String> {
        let mut topics = self.subscribed_topics.write().await;
        if !topics.contains(&topic.to_string()) {
            topics.push(topic.to_string());
        }
        Ok(())
    }
    
    async fn receive(&self) -> Result<Option<ConfigMessage>, String> {
        let queue = self.message_queue.read().await;
        let topics = self.subscribed_topics.read().await;
        let mut last_index = self.last_read_index.lock().await;
        
        // 从上次读取位置开始查找匹配的消息
        for (index, message) in queue.iter().enumerate().skip(*last_index) {
            if topics.iter().any(|topic| message.topic.starts_with(topic)) {
                *last_index = index + 1;
                return Ok(Some(message.clone()));
            }
        }
        
        // 更新读取位置到队列末尾
        *last_index = queue.len();
        Ok(None)
    }
    
    async fn unsubscribe(&self, topic: &str) -> Result<(), String> {
        let mut topics = self.subscribed_topics.write().await;
        topics.retain(|t| t != topic);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    
    #[tokio::test]
    async fn test_config_message_creation() {
        let message = ConfigMessage::new(
            "config.broadcast.config.update".to_string(),
            json!({"test": "data"}),
        )
        .with_header("event-type".to_string(), "config.update".to_string())
        .with_correlation_id("test-123".to_string());
        
        assert_eq!(message.topic, "config.broadcast.config.update");
        assert_eq!(message.payload, json!({"test": "data"}));
        assert_eq!(message.headers.get("event-type"), Some(&"config.update".to_string()));
        assert_eq!(message.correlation_id, Some("test-123".to_string()));
    }
    
    #[tokio::test]
    async fn test_in_memory_message_bus() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        
        // 创建发布者和订阅者
        let publisher = bus.create_publisher().await.unwrap();
        let subscriber = bus.create_subscriber().await.unwrap();
        
        // 订阅主题
        subscriber.subscribe("config.broadcast").await.unwrap();
        
        // 发布消息
        let message = ConfigMessage::new(
            "config.broadcast.config.update".to_string(),
            json!({"application": "test-service"}),
        );
        publisher.publish(message).await.unwrap();
        
        // 接收消息
        let received = subscriber.receive().await.unwrap();
        assert!(received.is_some());
        
        let msg = received.unwrap();
        assert_eq!(msg.topic, "config.broadcast.config.update");
        assert_eq!(msg.payload["application"], "test-service");
    }
    
    #[tokio::test]
    async fn test_abstract_config_broadcast_manager() {
        let bus = Arc::new(InMemoryConfigMessageBus::new());
        let manager = AbstractConfigBroadcastManager::new(
            bus,
            "test-service".to_string(),
            Some("config.broadcast".to_string()),
        );
        
        // 测试广播配置更新
        let result = manager.broadcast_config_update(
            "user-service",
            "prod",
            Some("v1.0"),
            "database.url",
            Some(json!("old-url")),
            json!("new-url"),
            Some("admin"),
        ).await;
        
        assert!(result.is_ok());
        
        // 测试获取统计信息
        let stats = manager.get_broadcast_stats().await;
        assert_eq!(stats.events_sent, 1);
    }
}